package com.zp.qualitypractice.rx;

import io.reactivex.Flowable;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;

/**
 * Created by izp on 2017/8/13.
 */

public class RxBusBackPressure {
        private final FlowableProcessor<Object> mBus;
//    private final Subject<Object> mBus;这个是非背压的

        private RxBusBackPressure() {
            // toSerialized method made bus thread safe
            mBus = PublishProcessor.create().toSerialized();
//        mBus = PublishSubject.create().toSerialized();这个是非背压的
        }

        public static RxBusBackPressure get() {
            return Holder.BUS;
        }

        public void post(Object obj) {
            mBus.onNext(obj);
        }

        public <T> Flowable<T> toFlowable(Class<T> tClass) {
            return mBus.ofType(tClass);
        }

        public Flowable<Object> toFlowable() {
            return mBus;
        }

        public boolean hasSubscribers() {
            return mBus.hasSubscribers();
        }

        private static class Holder {
            private static final RxBusBackPressure BUS = new RxBusBackPressure();
        }
}
